#include "proxy_invoke.h"
#include "proxy_message.h"
#include "log.h"
#include "util.h"

#include <pthread.h>
#include <stdint.h>  //uint8_t,uint16_t,uint32_t,uint64_t

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <ctype.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>


static thread_context * global_thd_cxt;
static pthread_mutex_t  global_lock;
static volatile int global_inited;

/**
 * lock pool
 * @param  pool [description]
 * @return      [description]
 */
static int cxt_lock() {
	return (pthread_mutex_lock(&global_lock) == 0) ? 1 : 0;
}

/**
 * unlock pool
 * @param  pool [description]
 * @return      [description]
 */
static int cxt_unlock() {
	return (pthread_mutex_unlock(&global_lock) == 0) ? 1 : 0;
}

static int init_thread_conext() {
	if (global_inited) {
		return 0;
	}
	global_thd_cxt = NULL;
	pthread_mutex_init(&global_lock, NULL);
	global_inited = 1;
	return 0;
}

static int fini_thread_context() {
	thread_context *current, *tmp;
	HASH_ITER(hh, global_thd_cxt, current, tmp) {
		HASH_DEL(global_thd_cxt, current);
		close(current->conn_fd);  //close fd
		free(current);
	}
	return 0;
}

static thread_context * find_thread_context(int thread_id) {
	thread_context * item = NULL;
	HASH_FIND_INT(global_thd_cxt, &thread_id, item);
	return item;
}

static int remove_fd_from_thread_context(int thread_id) {
	cxt_lock();
	thread_context * item = find_thread_context(thread_id);
	if (item) {
		HASH_DEL(global_thd_cxt, item);
		close(item->conn_fd);  //close fd
		INFO("disconnect success : %d", item->conn_fd );
		free(item);
	}
	cxt_unlock();
	return 0;
}

static int create_unix_socket(const char * proxy_sock_path) {
	int fd, t, len;
	struct sockaddr_un remote;
	if ((fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
		ERROR("create unix socket failed : %s, %s", proxy_sock_path, strerror(errno));
		return -1;
	}
	remote.sun_family = AF_UNIX;
	strcpy(remote.sun_path, proxy_sock_path);
	len = strlen(remote.sun_path) + sizeof(remote.sun_family);
	if (connect(fd, (struct sockaddr *)&remote, len) == -1) {
		ERROR("connect unix socket failed : %s, %s", proxy_sock_path, strerror(errno));
		return -1;
	}
	return fd;
}

static int create_tcp_socket(const char * host, int port) {
	int fd;
	int flag = 1;
	struct sockaddr_in pin;
	bzero(&pin, sizeof(pin));
	pin.sin_family = AF_INET;
	inet_pton(AF_INET, host, &pin.sin_addr);
	pin.sin_port = htons(port);
	fd = socket(AF_INET, SOCK_STREAM, 0);
	if (fd == -1) {
		ERROR("create tcp socket failed : %s:%d, %s", host, port, strerror(errno));
		return -1;
	}
	struct timeval timeout = {3, 0};
	setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (const char*)&timeout, sizeof(timeout));
	setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout, sizeof(timeout));
	int ret = connect(fd, (void *)&pin, sizeof(pin));
	if (ret) {
		ERROR("connect tcp socket failed : %s:%d, %s", host, port, strerror(errno));
		return -1;
	}
	return fd;
}

static int get_fd_from_thread_context(int thread_id, const char * proxy_sock_path) {
	cxt_lock();
	thread_context * item = find_thread_context(thread_id);
	if (item) {
		cxt_unlock();
		return item->conn_fd;
	}
	//create client fd
	int fd;
	char ip_buf[20];
	int port;
	if (parse_addr(proxy_sock_path, ip_buf, 20, &port) == 0) {
		fd = create_tcp_socket(ip_buf, port);
	} else {
		fd = create_unix_socket(proxy_sock_path);
	}
	if (fd < 0) {
		cxt_unlock();
		return -1;
	}
	//
	item = malloc(sizeof(thread_context));
	item->thread_id = thread_id;
	item->conn_fd = fd;
	HASH_ADD_INT(global_thd_cxt, thread_id, item);
	INFO("connect success : %d", fd );
	cxt_unlock();
	return fd;
}


/**
 * [invoke_proxy description]
 * @param  proxy_sock_path [description]
 * @param  group           [description]
 * @param  service         [description]
 * @param  version         [description]
 * @param  req_body        [description]
 * @param  req_len         [description]
 * @param  res_buf         [description]
 * @param  res_buf_len     [description]
 * @return                 [description]
 */
int invoke_proxy(const char * proxy_sock_path,
                 const char * group, const char * service, const char * version, const void * req_body, size_t req_len,
                 void ** res_buf, size_t * res_buf_len) {
	init_thread_conext();
	int curr_thread_id = pthread_self();
	int retry = 2;
	int success = 0;
	SoaProxyResponse * resp_msg = NULL;
	while (retry-- > 0) {
		int fd = get_fd_from_thread_context(curr_thread_id, proxy_sock_path);
		if (fd < 0) {
			break;
		}
		DEBUG("conn fd %d for thread %d", fd, curr_thread_id);
		if (proxy_request_send(fd, group, service, version, req_body, req_len)) {
			ERROR("proxy_request_send failed : %d, %s", fd, strerror(errno));
			remove_fd_from_thread_context(curr_thread_id);
			continue;
		}
		if (proxy_response_receive(fd, &resp_msg)) {
			ERROR("proxy_response_receive failed : %d, %s", fd, strerror(errno));
			remove_fd_from_thread_context(curr_thread_id);
			continue;
		}
		success = 1;
		break;
	}
	if (!success) {
		return -1;
	}
	if (resp_msg->status == 1) {
		ERROR("soa service invoke failed : %s,%s,%s", group, service, version);
		proxy_response_decode_clean(resp_msg);
		return -1;
	}
	if (resp_msg->response.len > 0) {
		*res_buf_len = resp_msg->response.len;
		*res_buf = malloc(resp_msg->response.len);
		memcpy(*res_buf, resp_msg->response.data, resp_msg->response.len);
	} else {
		*res_buf_len = 0;
		*res_buf = NULL;
	}
	proxy_response_decode_clean(resp_msg);
	return 0;
}
